package com.shujia.spark.mllib

import org.apache.spark.ml.linalg
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql._

object Demo5ImageData {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local[8]")
      .appName("person")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    import spark.implicits._

    /**
      * 1、读取赌片的数据
      *
      */

    val imageData: DataFrame = spark
      .read
      .format("image")
      .load("D:\\课件\\机器学习数据\\手写数字\\train")
      .select($"image.origin", $"image.data")

    imageData.printSchema()
    imageData.show(false)

    /**
      * 将二进制的图片转换向量
      *
      */
    val nameAndVector: DataFrame = imageData.map {
      case Row(origin: String, data: Array[Byte]) =>
        val comData: Array[Double] = data
          .map(byte => byte.toInt)
          //归一化，将数据转换成0或者1
          .map(i => {
          if (i >= 0) {
            0.0
          } else {
            1.0
          }
        })
        //将特征转换成向量
        val vector: linalg.Vector = Vectors.dense(comData)

        //取出图片的名称
        val name: String = origin.split("/").last

        (name, vector)
    }.toDF("name", "features")

    /**
      * 读取标签数据
      *
      */

    val nameAndLabel: DataFrame = spark
      .read
      .format("csv")
      .option("sep", " ")
      .schema("name STRING, label DOUBLE")
      .load("D:\\课件\\机器学习数据\\手写数字\\train.txt")

    /**
      * 关联得到图片的标签
      *
      */

    val joinDF: DataFrame = nameAndVector.join(nameAndLabel, "name")

    //取出目标值和特征值
    val data: DataFrame = joinDF.select("label", "features")

    //将处理好的数据保存为SM格式
    data
      .write
      .format("libsvm")
      .mode(SaveMode.Overwrite)
      .save("data/image_data")


  }

}
